#!/bin/bash
# set -x
KAFKA_HOME=/home/k2data/kafka_2.10-0.8.2.2
while true; do
	# for raw-binary input
	raw_binary_processed_offset=$(echo "get /zk_test/dataplatform/storm/binary-kafka-spout/partition_0" | $KAFKA_HOME/bin/zookeeper-shell.sh kmxnode6,kmxnode4,kmxnode3 2>/dev/null | tail -n 1 | jq '.offset')
	echo "put sdlg.kafka.raw_binary.offset.processed $(date +%s) $raw_binary_processed_offset host=$(hostname -s)"
	raw_binary_latest_offset=$($KAFKA_HOME/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.1.88:19092,192.168.1.89:19092 --time -1 --topic defaultUser-defaultSpace-raw-binary | grep :0: | awk -F [:] '{print $3}')
	echo "put sdlg.kafka.raw_binary.offset.latest $(date +%s) $raw_binary_latest_offset host=$(hostname -s)"
	# for raw input
	raw_processed_offset=$(echo "get /zk_test/dataplatform/storm/Fixed-format-kafka-spout/partition_0" | $KAFKA_HOME/bin/zookeeper-shell.sh kmxnode6,kmxnode4,kmxnode3 2>/dev/null | tail -n 1 | jq '.offset')
	echo "put sdlg.kafka.raw.offset.processed $(date +%s) $raw_processed_offset host=$(hostname -s)"
	raw_latest_offset=$($KAFKA_HOME/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.1.88:19092,192.168.1.89:19092  --time -1 --topic defaultUser-defaultSpace-raw | grep :0: | awk -F [:] '{print $3}')
	echo "put sdlg.kafka.raw.offset.latest $(date +%s) $raw_latest_offset host=$(hostname -s)"
	sleep 10
done #| nc 211.154.7.134 4242
